//! # 配置监听器模块
//!
//! 提供配置变更的实时监听功能，支持多种监听策略和事件通知机制。
//!
//! ## 核心功能
//!
//! * **轮询监听** - 定期检查配置变更
//! * **推送监听** - 支持服务器主动推送变更通知
//! * **事件过滤** - 支持按应用、环境等维度过滤事件
//! * **批量监听** - 同时监听多个配置的变更
//! * **延迟通知** - 防抖动机制，避免频繁变更通知
//! * **监听恢复** - 监听中断后自动恢复

use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, SystemTime};

use async_trait::async_trait;
use tokio::sync::{broadcast, RwLock};
use tokio::time::{interval, Interval};
use tracing::{debug, error, info, warn};
use serde::{Deserialize, Serialize};
use futures_util::{SinkExt, StreamExt};
use uuid::Uuid;

use crate::core::*;
use crate::RustCloudConfig;
use crate::client::{
    ConfigChangeEvent, ConfigEventType, ConfigChangeListener,
    RemoteConfigFetcher, ConfigClientResult as ClientResult,
    ConfigClientError, ConfigCache
};

/// SSE事件流提供者
pub struct SseEventStreamProvider {
    client: reqwest::Client,
}

impl SseEventStreamProvider {
    pub fn new() -> Self {
        Self {
            client: reqwest::Client::new(),
        }
    }
}

#[async_trait]
impl EventStreamProvider for SseEventStreamProvider {
    async fn create_stream(&self, endpoint: &str, options: &StreamOptions) -> ConfigurationResult<Box<dyn EventStream>> {
        if options.stream_type != StreamType::ServerSentEvents {
            return Err("Unsupported stream type for SSE provider".into());
        }
        
        let sse_stream = SseEventStreamImpl::new(&self.client, endpoint, options.clone()).await?;
        Ok(Box::new(sse_stream))
    }
    
    fn supported_stream_types(&self) -> Vec<StreamType> {
        vec![StreamType::ServerSentEvents]
    }
    
    async fn validate_endpoint(&self, endpoint: &str) -> ConfigurationResult<bool> {
        let url = if endpoint.starts_with("http://") || endpoint.starts_with("https://") {
            format!("{}/health", endpoint)
        } else {
            format!("http://{}/health", endpoint)
        };
        
        match self.client.get(&url).send().await {
            Ok(response) => Ok(response.status().is_success()),
            Err(_) => Ok(false),
        }
    }
}

/// WebSocket事件流提供者
pub struct WebSocketEventStreamProvider;

impl WebSocketEventStreamProvider {
    pub fn new() -> Self {
        Self
    }
}

#[async_trait]
impl EventStreamProvider for WebSocketEventStreamProvider {
    async fn create_stream(&self, endpoint: &str, options: &StreamOptions) -> ConfigurationResult<Box<dyn EventStream>> {
        if options.stream_type != StreamType::WebSocket {
            return Err("Unsupported stream type for WebSocket provider".into());
        }
        
        let ws_stream = WebSocketEventStreamImpl::new(endpoint, options.clone()).await?;
        Ok(Box::new(ws_stream))
    }
    
    fn supported_stream_types(&self) -> Vec<StreamType> {
        vec![StreamType::WebSocket]
    }
    
    async fn validate_endpoint(&self, endpoint: &str) -> ConfigurationResult<bool> {
        // 模拟WebSocket端点验证
        Ok(endpoint.starts_with("ws://") || endpoint.starts_with("wss://") || 
           endpoint.contains("://"))
    }
}

/// SSE事件流实现
pub struct SseEventStreamImpl {
    endpoint: String,
    options: StreamOptions,
    status: StreamStatus,
    client: reqwest::Client,
}

impl SseEventStreamImpl {
    async fn new(client: &reqwest::Client, endpoint: &str, options: StreamOptions) -> ConfigurationResult<Self> {
        Ok(Self {
            endpoint: endpoint.to_string(),
            options,
            status: StreamStatus::Disconnected,
            client: client.clone(),
        })
    }
}

#[async_trait]
impl EventStream for SseEventStreamImpl {
    async fn subscribe(&mut self, environments: &[ConfigEnvironment]) -> ConfigurationResult<()> {
        info!("📡 SSE订阅环境: {:?}", environments);
        self.status = StreamStatus::Connected;
        Ok(())
    }
    
    async fn receive_event(&mut self) -> ConfigurationResult<Option<StreamEvent>> {
        // 模拟SSE事件接收
        Ok(Some(StreamEvent {
            event_type: "config_update".to_string(),
            data: serde_json::json!({"message": "test event"}),
            event_id: Some(Uuid::new_v4().to_string()),
            timestamp: SystemTime::now(),
            retry: Some(1000),
        }))
    }
    
    async fn send_heartbeat(&mut self) -> ConfigurationResult<()> {
        debug!("💓 发送SSE心跳");
        Ok(())
    }
    
    async fn close(&mut self) -> ConfigurationResult<()> {
        self.status = StreamStatus::Disconnected;
        info!("🔌 SSE连接已关闭");
        Ok(())
    }
    
    fn get_status(&self) -> StreamStatus {
        self.status.clone()
    }
}

/// WebSocket事件流实现
pub struct WebSocketEventStreamImpl {
    endpoint: String,
    options: StreamOptions,
    status: StreamStatus,
}

impl WebSocketEventStreamImpl {
    async fn new(endpoint: &str, options: StreamOptions) -> ConfigurationResult<Self> {
        Ok(Self {
            endpoint: endpoint.to_string(),
            options,
            status: StreamStatus::Disconnected,
        })
    }
}

#[async_trait]
impl EventStream for WebSocketEventStreamImpl {
    async fn subscribe(&mut self, environments: &[ConfigEnvironment]) -> ConfigurationResult<()> {
        info!("🔌 WebSocket订阅环境: {:?}", environments);
        self.status = StreamStatus::Connected;
        Ok(())
    }
    
    async fn receive_event(&mut self) -> ConfigurationResult<Option<StreamEvent>> {
        // 模拟WebSocket事件接收
        Ok(Some(StreamEvent {
            event_type: "config_update".to_string(),
            data: serde_json::json!({"message": "websocket test event"}),
            event_id: Some(Uuid::new_v4().to_string()),
            timestamp: SystemTime::now(),
            retry: None,
        }))
    }
    
    async fn send_heartbeat(&mut self) -> ConfigurationResult<()> {
        debug!("🏓 发送WebSocket心跳");
        Ok(())
    }
    
    async fn close(&mut self) -> ConfigurationResult<()> {
        self.status = StreamStatus::Disconnected;
        info!("🔌 WebSocket连接已关闭");
        Ok(())
    }
    
    fn get_status(&self) -> StreamStatus {
        self.status.clone()
    }
}

/// 配置监听器状态
#[derive(Debug, Clone, PartialEq)]
pub enum WatcherState {
    /// 停止状态
    Stopped,
    /// 运行状态
    Running,
    /// 暂停状态
    Paused,
    /// 错误状态
    Error(String),
}

/// 监听策略
#[derive(Debug, Clone)]
pub enum WatchStrategy {
    /// 轮询策略
    Polling {
        /// 轮询间隔
        interval: Duration,
        /// 版本比较
        version_check: bool,
    },
    /// WebSocket推送策略
    WebSocketPush {
        /// 推送端点
        endpoint: String,
        /// 重连间隔
        reconnect_interval: Duration,
    },
    /// SSE推送策略（Server-Sent Events）
    SsePush {
        /// SSE端点
        endpoint: String,
        /// 重连间隔
        reconnect_interval: Duration,
        /// 心跳间隔
        heartbeat_interval: Duration,
    },
    /// 混合策略
    Hybrid {
        /// 主要使用推送，fallback到轮询
        push_endpoint: String,
        polling_interval: Duration,
        /// 推送类型（WebSocket或SSE）
        push_type: PushType,
    },
}

/// 推送类型
#[derive(Debug, Clone)]
pub enum PushType {
    WebSocket,
    Sse,
}

impl Default for WatchStrategy {
    fn default() -> Self {
        Self::Polling {
            interval: Duration::from_secs(30),
            version_check: true,
        }
    }
}

/// 配置监听器
pub struct ConfigWatcher {
    /// 监听的环境列表
    environments: Arc<RwLock<Vec<ConfigEnvironment>>>,
    /// 远程配置获取器
    fetcher: Arc<dyn RemoteConfigFetcher>,
    /// 监听策略
    strategy: WatchStrategy,
    /// 当前状态
    state: Arc<RwLock<WatcherState>>,
    /// 配置版本缓存
    version_cache: Arc<RwLock<HashMap<String, String>>>,
    /// 配置内容缓存
    config_cache: Arc<RwLock<HashMap<String, RustCloudConfig>>>,
    /// 事件发送器
    event_sender: broadcast::Sender<ConfigChangeEvent>,
    /// 延迟通知时间
    debounce_duration: Duration,
    /// 最后通知时间
    last_notification: Arc<RwLock<HashMap<String, SystemTime>>>,
}

impl ConfigWatcher {
    /// 创建新的配置监听器
    pub fn new(
        fetcher: Arc<dyn RemoteConfigFetcher>,
        strategy: WatchStrategy,
        event_sender: broadcast::Sender<ConfigChangeEvent>,
    ) -> Self {
        Self {
            environments: Arc::new(RwLock::new(Vec::new())),
            fetcher,
            strategy,
            state: Arc::new(RwLock::new(WatcherState::Stopped)),
            version_cache: Arc::new(RwLock::new(HashMap::new())),
            config_cache: Arc::new(RwLock::new(HashMap::new())),
            event_sender,
            debounce_duration: Duration::from_millis(500), // 500ms防抖
            last_notification: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// 添加监听环境
    pub async fn add_environment(&self, environment: ConfigEnvironment) {
        let mut environments = self.environments.write().await;
        if !environments.contains(&environment) {
            environments.push(environment.clone());
            info!("👂 添加监听环境: {}:{}:{}", 
                environment.application, environment.profile, environment.label);
        }
    }

    /// 移除监听环境
    pub async fn remove_environment(&self, environment: &ConfigEnvironment) {
        let mut environments = self.environments.write().await;
        environments.retain(|env| env != environment);
        info!("🗑️ 移除监听环境: {}:{}:{}", 
            environment.application, environment.profile, environment.label);
    }

    /// 开始监听
    pub async fn start(&self) -> ClientResult<()> {
        let mut state = self.state.write().await;
        
        match *state {
            WatcherState::Running => {
                debug!("⚠️ 监听器已在运行中");
                return Ok(());
            }
            _ => {
                *state = WatcherState::Running;
            }
        }
        drop(state);

        info!("🚀 启动配置监听器");

        // 根据策略启动不同的监听任务
        match &self.strategy {
            WatchStrategy::Polling { interval, version_check } => {
                self.start_polling_watch(*interval, *version_check).await;
            }
            WatchStrategy::WebSocketPush { endpoint, reconnect_interval } => {
                self.start_websocket_push_watch(endpoint.clone(), *reconnect_interval).await;
            }
            WatchStrategy::SsePush { endpoint, reconnect_interval, heartbeat_interval } => {
                self.start_sse_push_watch(endpoint.clone(), *reconnect_interval, *heartbeat_interval).await;
            }
            WatchStrategy::Hybrid { push_endpoint, polling_interval, push_type } => {
                self.start_hybrid_watch(push_endpoint.clone(), *polling_interval).await;
            }
        }

        Ok(())
    }

    /// 停止监听
    pub async fn stop(&self) {
        let mut state = self.state.write().await;
        *state = WatcherState::Stopped;
        info!("🛑 停止配置监听器");
    }

    /// 暂停监听
    pub async fn pause(&self) {
        let mut state = self.state.write().await;
        if *state == WatcherState::Running {
            *state = WatcherState::Paused;
            info!("⏸️ 暂停配置监听器");
        }
    }

    /// 恢复监听
    pub async fn resume(&self) {
        let mut state = self.state.write().await;
        if *state == WatcherState::Paused {
            *state = WatcherState::Running;
            info!("▶️ 恢复配置监听器");
        }
    }

    /// 获取监听器状态
    pub async fn get_state(&self) -> WatcherState {
        self.state.read().await.clone()
    }

    /// 启动轮询监听
    async fn start_polling_watch(&self, poll_interval: Duration, version_check: bool) {
        let environments = self.environments.clone();
        let fetcher = self.fetcher.clone();
        let state = self.state.clone();
        let version_cache = self.version_cache.clone();
        let config_cache = self.config_cache.clone();
        let event_sender = self.event_sender.clone();
        let debounce_duration = self.debounce_duration;
        let last_notification = self.last_notification.clone();

        tokio::spawn(async move {
            let mut ticker = interval(poll_interval);
            debug!("📊 开始轮询监听，间隔: {:?}", poll_interval);

            loop {
                ticker.tick().await;

                // 检查是否应该继续运行
                let current_state = state.read().await.clone();
                match current_state {
                    WatcherState::Stopped => {
                        debug!("🛑 监听器已停止，退出轮询任务");
                        break;
                    }
                    WatcherState::Paused => {
                        debug!("⏸️ 监听器已暂停，跳过本次轮询");
                        continue;
                    }
                    WatcherState::Error(_) => {
                        warn!("❌ 监听器处于错误状态，跳过本次轮询");
                        continue;
                    }
                    WatcherState::Running => {
                        // 继续执行
                    }
                }

                let envs = environments.read().await.clone();
                for environment in envs {
                    if let Err(e) = Self::check_config_change(
                        &environment,
                        &fetcher,
                        version_check,
                        &version_cache,
                        &config_cache,
                        &event_sender,
                        debounce_duration,
                        &last_notification,
                    ).await {
                        error!("❌ 检查配置变更失败 {}:{}:{}: {}", 
                            environment.application, environment.profile, environment.label, e);
                    }
                }
            }
        });
    }

    /// 启动推送监听
    async fn start_push_watch(&self, endpoint: String, reconnect_interval: Duration) {
        let environments = self.environments.clone();
        let state = self.state.clone();
        let config_cache = self.config_cache.clone();
        let event_sender = self.event_sender.clone();
        let last_notification = self.last_notification.clone();
        let debounce_duration = self.debounce_duration;

        tokio::spawn(async move {
            info!("🔌 开始WebSocket推送监听，端点: {}", endpoint);
            
            loop {
                // 检查是否应该继续运行
                let current_state = state.read().await.clone();
                match current_state {
                    WatcherState::Stopped => {
                        debug!("🛑 监听器已停止，退出推送任务");
                        break;
                    }
                    WatcherState::Paused => {
                        debug!("⏸️ 监听器已暂停，等待恢复");
                        tokio::time::sleep(Duration::from_secs(1)).await;
                        continue;
                    }
                    WatcherState::Error(_) => {
                        warn!("❌ 监听器处于错误状态，尝试重连");
                        tokio::time::sleep(reconnect_interval).await;
                        continue;
                    }
                    WatcherState::Running => {
                        // 继续执行
                    }
                }

                match Self::establish_websocket_connection(&endpoint).await {
                    Ok(mut ws_stream) => {
                        info!("✅ WebSocket连接已建立");
                        
                        // 发送订阅消息
                        let envs = environments.read().await.clone();
                        if let Err(e) = Self::send_subscription_message(&mut ws_stream, &envs).await {
                            error!("❌ 发送订阅消息失败: {}", e);
                            continue;
                        }

                        // 监听消息
                        loop {
                            match Self::receive_websocket_message(&mut ws_stream).await {
                                Ok(Some(message)) => {
                                    if let Err(e) = Self::process_push_message(
                                        message,
                                        &config_cache,
                                        &event_sender,
                                        debounce_duration,
                                        &last_notification,
                                    ).await {
                                        error!("❌ 处理推送消息失败: {}", e);
                                    }
                                }
                                Ok(None) => {
                                    debug!("🔌 WebSocket连接已关闭");
                                    break;
                                }
                                Err(e) => {
                                    error!("❌ WebSocket消息接收失败: {}", e);
                                    break;
                                }
                            }
                            
                            // 检查监听器状态
                            let current_state = state.read().await.clone();
                            if !matches!(current_state, WatcherState::Running) {
                                break;
                            }
                        }
                    }
                    Err(e) => {
                        error!("❌ WebSocket连接失败: {}，{}秒后重试", e, reconnect_interval.as_secs());
                        
                        // 更新状态为错误
                        let mut state_write = state.write().await;
                        *state_write = WatcherState::Error(e.to_string());
                        drop(state_write);
                        
                        tokio::time::sleep(reconnect_interval).await;
                        
                        // 重置状态为运行
                        let mut state_write = state.write().await;
                        if matches!(*state_write, WatcherState::Error(_)) {
                            *state_write = WatcherState::Running;
                        }
                    }
                }
            }
        });
    }

    /// 启动混合监听
    async fn start_hybrid_watch(&self, push_endpoint: String, polling_interval: Duration) {
        info!("🔄 开始混合监听模式 - WebSocket推送 + 轮询备份");
        
        // 同时启动推送监听和轮询监听作为备份
        self.start_push_watch(push_endpoint, Duration::from_secs(30)).await;
        
        // 启动轮询作为备份，间隔更长以减少服务器压力
        let backup_interval = polling_interval * 3; // 备份轮询间隔是主轮询间隔的3倍
        self.start_polling_watch(backup_interval, true).await;
        
        info!("✅ 混合监听模式已启动");
    }

    /// 建立 WebSocket 连接
    async fn establish_websocket_connection(endpoint: &str) -> ClientResult<WebSocketStream> {
        use tokio_tungstenite::{connect_async, tungstenite::protocol::Message};
        
        debug!("🔌 尝试建立WebSocket连接: {}", endpoint);
        
        // 构建WebSocket URL
        let ws_url = if endpoint.starts_with("ws://") || endpoint.starts_with("wss://") {
            endpoint.to_string()
        } else {
            format!("ws://{}/ws/config-push", endpoint)
        };
        
        match connect_async(&ws_url).await {
            Ok((ws_stream, response)) => {
                info!("✅ WebSocket连接成功: {} (状态: {})", ws_url, response.status());
                Ok(ws_stream)
            }
            Err(e) => {
                error!("❌ WebSocket连接失败: {} - {}", ws_url, e);
                Err(ConfigurationClientError::ConfigFetchError(format!("WebSocket连接失败: {}", e)))
            }
        }
    }

    /// 发送订阅消息
    async fn send_subscription_message(
        ws_stream: &mut WebSocketStream,
        environments: &[ConfigEnvironment],
    ) -> ClientResult<()> {
        use tokio_tungstenite::tungstenite::protocol::Message;
        use futures_util::SinkExt;
        
        let subscription_message = SubscriptionMessage {
            action: "subscribe".to_string(),
            environments: environments.to_vec(),
        };
        
        let message_json = serde_json::to_string(&subscription_message)
            .map_err(|e| ConfigurationClientError::SerializationError {
                source: serde_json::Error::io(std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))
            })?;
        
        ws_stream.send(Message::Text(message_json)).await
            .map_err(|e| ConfigurationClientError::ConfigFetchError(format!("发送订阅消息失败: {}", e)))?;
        
        info!("📨 已发送订阅消息，监听 {} 个环境", environments.len());
        Ok(())
    }

    /// 接收 WebSocket 消息
    async fn receive_websocket_message(
        ws_stream: &mut WebSocketStream,
    ) -> ClientResult<Option<PushMessage>> {
        use tokio_tungstenite::tungstenite::protocol::Message;
        use futures_util::StreamExt;
        
        match ws_stream.next().await {
            Some(Ok(message)) => {
                match message {
                    Message::Text(text) => {
                        debug!("📥 收到WebSocket文本消息: {}", text);
                        
                        let push_message: PushMessage = serde_json::from_str(&text)
                            .map_err(|e| ConfigurationClientError::SerializationError {
                                source: e
                            })?;
                        
                        Ok(Some(push_message))
                    }
                    Message::Binary(data) => {
                        debug!("📥 收到WebSocket二进制消息，长度: {}", data.len());
                        
                        let text = String::from_utf8(data)
                            .map_err(|e| ConfigurationClientError::SerializationError {
                                source: serde_json::Error::io(std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))
                            })?;
                        
                        let push_message: PushMessage = serde_json::from_str(&text)
                            .map_err(|e| ConfigurationClientError::SerializationError {
                                source: e
                            })?;
                        
                        Ok(Some(push_message))
                    }
                    Message::Ping(data) => {
                        debug!("🏓 收到Ping消息，发送Pong回复");
                        ws_stream.send(Message::Pong(data)).await
                            .map_err(|e| ConfigurationClientError::ConfigFetchError(format!("发送Pong失败: {}", e)))?;
                        Ok(None)
                    }
                    Message::Pong(_) => {
                        debug!("🏓 收到Pong回复");
                        Ok(None)
                    }
                    Message::Close(close_frame) => {
                        info!("🔌 WebSocket连接关闭: {:?}", close_frame);
                        Ok(None)
                    }
                    _ => {
                        warn!("⚠️ 收到未知类型的WebSocket消息");
                        Ok(None)
                    }
                }
            }
            Some(Err(e)) => {
                error!("❌ WebSocket消息接收错误: {}", e);
                Err(ConfigurationClientError::ConfigFetchError(format!("WebSocket接收错误: {}", e)))
            }
            None => {
                debug!("🔌 WebSocket流已结束");
                Ok(None)
            }
        }
    }

    /// 处理推送消息
    async fn process_push_message(
        message: PushMessage,
        config_cache: &Arc<RwLock<HashMap<String, RustCloudConfig>>>,
        event_sender: &broadcast::Sender<ConfigChangeEvent>,
        debounce_duration: Duration,
        last_notification: &Arc<RwLock<HashMap<String, SystemTime>>>,
    ) -> ClientResult<()> {
        let env_key = format!("{}:{}:{}", 
            message.environment.application, 
            message.environment.profile, 
            message.environment.label
        );
        
        info!("📨 处理推送消息: {} (类型: {})", env_key, message.message_type);
        
        // 检查防抖
        {
            let notifications = last_notification.read().await;
            if let Some(last_time) = notifications.get(&env_key) {
                if last_time.elapsed().unwrap_or_default() < debounce_duration {
                    debug!("⏰ 防抖期内，跳过推送通知: {}", env_key);
                    return Ok(());
                }
            }
        }
        
        match message.message_type.as_str() {
            "config_updated" | "config_created" => {
                if let Some(ref new_config) = message.config {
                    // 更新缓存
                    {
                        let mut cache = config_cache.write().await;
                        cache.insert(env_key.clone(), new_config.clone());
                    }
                    
                    // 发送变更事件
                    let event = ConfigChangeEvent {
                        environment: message.environment.clone(),
                        event_type: if message.message_type == "config_created" {
                            ConfigEventType::Created
                        } else {
                            ConfigEventType::Updated
                        },
                        old_config: None,
                        new_config: Some(new_config.clone()),
                        timestamp: message.timestamp,
                    };
                    
                    if let Err(e) = event_sender.send(event) {
                        warn!("⚠️ 发送配置变更事件失败: {}", e);
                    }
                    
                    info!("✅ 配置已更新: {}", env_key);
                } else {
                    warn!("⚠️ 推送消息缺少配置数据: {}", env_key);
                }
            }
            "config_deleted" => {
                // 从缓存中删除
                let old_config = {
                    let mut cache = config_cache.write().await;
                    cache.remove(&env_key)
                };
                
                // 发送删除事件
                let event = ConfigChangeEvent {
                    environment: message.environment.clone(),
                    event_type: ConfigEventType::Deleted,
                    old_config,
                    new_config: None,
                    timestamp: message.timestamp,
                };
                
                if let Err(e) = event_sender.send(event) {
                    warn!("⚠️ 发送配置删除事件失败: {}", e);
                }
                
                info!("🗑️ 配置已删除: {}", env_key);
            }
            "heartbeat" => {
                debug!("💓 收到心跳消息: {}", env_key);
            }
            _ => {
                warn!("⚠️ 未知的推送消息类型: {}", message.message_type);
            }
        }
        
        // 更新最后通知时间
        {
            let mut notifications = last_notification.write().await;
            notifications.insert(env_key, SystemTime::now());
        }
        
        Ok(())
    }

    /// 检查配置变更
    async fn check_config_change(
        environment: &ConfigEnvironment,
        fetcher: &Arc<dyn RemoteConfigFetcher>,
        version_check: bool,
        version_cache: &Arc<RwLock<HashMap<String, String>>>,
        config_cache: &Arc<RwLock<HashMap<String, RustCloudConfig>>>,
        event_sender: &broadcast::Sender<ConfigChangeEvent>,
        debounce_duration: Duration,
        last_notification: &Arc<RwLock<HashMap<String, SystemTime>>>,
    ) -> ClientResult<()> {
        let env_key = format!("{}:{}:{}", 
            environment.application, environment.profile, environment.label);

        debug!("🔍 检查配置变更: {}", env_key);

        // 检查是否在防抖期内
        {
            let notifications = last_notification.read().await;
            if let Some(last_time) = notifications.get(&env_key) {
                if last_time.elapsed().unwrap_or_default() < debounce_duration {
                    debug!("⏰ 防抖期内，跳过通知: {}", env_key);
                    return Ok(());
                }
            }
        }

        let mut has_changed = false;
        let mut old_config = None;
        let mut new_config = None;

        if version_check {
            // 基于版本的变更检测
            match fetcher.get_version(environment).await {
                Ok(current_version) => {
                    let mut cache = version_cache.write().await;
                    let cached_version = cache.get(&env_key);
                    
                    if cached_version.map(|v| v != &current_version).unwrap_or(true) {
                        debug!("📝 检测到版本变更: {} -> {}", 
                            cached_version.unwrap_or(&"none".to_string()), current_version);
                        
                        // 获取新配置
                        match fetcher.fetch_config(environment).await {
                            Ok(config) => {
                                // 获取旧配置
                                let config_cache_read = config_cache.read().await;
                                old_config = config_cache_read.get(&env_key).cloned();
                                drop(config_cache_read);

                                // 更新缓存
                                cache.insert(env_key.clone(), current_version);
                                drop(cache);

                                let mut config_cache_write = config_cache.write().await;
                                config_cache_write.insert(env_key.clone(), config.clone());
                                drop(config_cache_write);

                                new_config = Some(config);
                                has_changed = true;
                            }
                            Err(e) => {
                                warn!("⚠️ 获取新配置失败: {}", e);
                                return Err(e);
                            }
                        }
                    }
                }
                Err(e) => {
                    warn!("⚠️ 获取版本信息失败: {}", e);
                    // 版本检查失败时，直接检查配置内容
                    return Self::check_config_content_change(
                        environment,
                        fetcher,
                        &env_key,
                        config_cache,
                        event_sender,
                        last_notification,
                    ).await;
                }
            }
        } else {
            // 基于内容的变更检测
            return Self::check_config_content_change(
                environment,
                fetcher,
                &env_key,
                config_cache,
                event_sender,
                last_notification,
            ).await;
        }

        // 发送变更事件
        if has_changed {
            let event = ConfigChangeEvent {
                event_type: if old_config.is_some() { 
                    ConfigEventType::Updated 
                } else { 
                    ConfigEventType::Created 
                },
                environment: environment.clone(),
                new_config,
                old_config,
                timestamp: SystemTime::now(),
            };

            if let Err(e) = event_sender.send(event) {
                warn!("⚠️ 发送配置变更事件失败: {}", e);
            } else {
                info!("📢 配置变更事件已发送: {}", env_key);
                
                // 更新最后通知时间
                let mut notifications = last_notification.write().await;
                notifications.insert(env_key, SystemTime::now());
            }
        }

        Ok(())
    }

    /// 检查配置内容变更
    async fn check_config_content_change(
        environment: &ConfigEnvironment,
        fetcher: &Arc<dyn RemoteConfigFetcher>,
        env_key: &str,
        config_cache: &Arc<RwLock<HashMap<String, RustCloudConfig>>>,
        event_sender: &broadcast::Sender<ConfigChangeEvent>,
        last_notification: &Arc<RwLock<HashMap<String, SystemTime>>>,
    ) -> ClientResult<()> {
        match fetcher.fetch_config(environment).await {
            Ok(current_config) => {
                let mut cache = config_cache.write().await;
                let cached_config = cache.get(env_key);
                
                // 简单的配置比较（实际应该实现更精确的比较）
                let has_changed = match cached_config {
                    Some(cached) => {
                        // 这里应该实现更精确的配置比较逻辑
                        serde_json::to_string(cached).unwrap_or_default() != 
                        serde_json::to_string(&current_config).unwrap_or_default()
                    }
                    None => true,
                };

                if has_changed {
                    debug!("📝 检测到配置内容变更: {}", env_key);
                    
                    let old_config = cached_config.cloned();
                    cache.insert(env_key.to_string(), current_config.clone());
                    drop(cache);

                    let event = ConfigChangeEvent {
                        event_type: if old_config.is_some() { 
                            ConfigEventType::Updated 
                        } else { 
                            ConfigEventType::Created 
                        },
                        environment: environment.clone(),
                        new_config: Some(current_config),
                        old_config,
                        timestamp: SystemTime::now(),
                    };

                    if let Err(e) = event_sender.send(event) {
                        warn!("⚠️ 发送配置变更事件失败: {}", e);
                    } else {
                        info!("📢 配置变更事件已发送: {}", env_key);
                        
                        // 更新最后通知时间
                        let mut notifications = last_notification.write().await;
                        notifications.insert(env_key.to_string(), SystemTime::now());
                    }
                }
            }
            Err(e) => {
                warn!("⚠️ 获取配置内容失败: {}", e);
                return Err(e);
            }
        }

        Ok(())
    }

    /// 手动触发配置检查
    pub async fn trigger_check(&self) -> ClientResult<()> {
        let environments = self.environments.read().await.clone();
        
        for environment in environments {
            Self::check_config_change(
                &environment,
                &self.fetcher,
                true, // 强制版本检查
                &self.version_cache,
                &self.config_cache,
                &self.event_sender,
                Duration::from_millis(0), // 跳过防抖
                &self.last_notification,
            ).await?;
        }

        info!("🔄 手动配置检查完成");
        Ok(())
    }
    
    /// 启动WebSocket推送监听
    async fn start_websocket_push_watch(&self, endpoint: String, reconnect_interval: Duration) {
        let environments = self.environments.clone();
        let state = self.state.clone();
        let event_sender = self.event_sender.clone();
        let config_cache = self.config_cache.clone();
        let last_notification = self.last_notification.clone();
        
        tokio::spawn(async move {
            let mut reconnect_count = 0;
            
            loop {
                // 检查是否应该停止
                {
                    let current_state = state.read().await;
                    if *current_state == WatcherState::Stopped {
                        break;
                    }
                    if *current_state == WatcherState::Paused {
                        tokio::time::sleep(Duration::from_secs(1)).await;
                        continue;
                    }
                }
                
                // 建立WebSocket连接
                match Self::establish_websocket_connection(&endpoint).await {
                    Ok(mut ws_stream) => {
                        info!("🔌 WebSocket 连接已建立: {}", endpoint);
                        reconnect_count = 0;
                        
                        // 发送订阅消息
                        let envs = environments.read().await.clone();
                        if let Err(e) = Self::send_subscription_message(&mut ws_stream, &envs).await {
                            error!("❌ 发送订阅消息失败: {}", e);
                            continue;
                        }
                        
                        // 消息循环
                        loop {
                            match Self::receive_websocket_message(&mut ws_stream).await {
                                Ok(Some(message)) => {
                                    if let Err(e) = Self::process_push_message(
                                        message,
                                        &config_cache,
                                        &event_sender,
                                        Duration::from_secs(30), // debounce_duration
                                        &last_notification,
                                    ).await {
                                        warn!("⚠️ 处理推送消息失败: {}", e);
                                    }
                                }
                                Ok(None) => {
                                    // 连接可能已关闭，准备重连
                                    break;
                                }
                                Err(e) => {
                                    error!("❌ WebSocket 消息接收错误: {}", e);
                                    break;
                                }
                            }
                        }
                    }
                    Err(e) => {
                        error!("❌ WebSocket 连接失败: {}", e);
                        reconnect_count += 1;
                    }
                }
                
                // 等待重连
                let wait_time = std::cmp::min(
                    reconnect_interval.as_secs() * 2_u64.pow(reconnect_count.min(5)),
                    300, // 最大5分钟
                );
                
                warn!("⏰ {} 秒后尝试重新连接... (第{}次重试)", wait_time, reconnect_count + 1);
                tokio::time::sleep(Duration::from_secs(wait_time)).await;
            }
            
            info!("🛑 WebSocket 推送监听已停止");
        });
    }
    
    /// 启动SSE推送监听
    async fn start_sse_push_watch(&self, endpoint: String, reconnect_interval: Duration, heartbeat_interval: Duration) {
        let environments = self.environments.clone();
        let state = self.state.clone();
        let event_sender = self.event_sender.clone();
        let config_cache = self.config_cache.clone();
        let last_notification = self.last_notification.clone();
        
        tokio::spawn(async move {
            let mut reconnect_count = 0;
            
            loop {
                // 检查是否应该停止
                {
                    let current_state = state.read().await;
                    if *current_state == WatcherState::Stopped {
                        break;
                    }
                    if *current_state == WatcherState::Paused {
                        tokio::time::sleep(Duration::from_secs(1)).await;
                        continue;
                    }
                }
                
                // 建立SSE连接
                match Self::establish_sse_connection(&endpoint, &environments).await {
                    Ok(mut event_stream) => {
                        info!("📡 SSE 连接已建立: {}", endpoint);
                        reconnect_count = 0;
                        
                        // 设置心跳检查
                        let mut heartbeat_timer = interval(heartbeat_interval);
                        let mut last_heartbeat = SystemTime::now();
                        
                        // 消息循环
                        loop {
                            tokio::select! {
                                // 接收SSE事件
                                event_result = Self::receive_sse_event(&mut event_stream) => {
                                    match event_result {
                                        Ok(Some(message)) => {
                                            last_heartbeat = SystemTime::now();
                                            if let Err(e) = Self::process_push_message(
                                                message,
                                                &config_cache,
                                                &event_sender,
                                                debounce_duration,
                                                &last_notification,
                                            ).await {
                                                warn!("⚠️ 处理SSE推送消息失败: {}", e);
                                            }
                                        }
                                        Ok(None) => {
                                            // 连接可能已关闭，准备重连
                                            break;
                                        }
                                        Err(e) => {
                                            error!("❌ SSE 事件接收错误: {}", e);
                                            break;
                                        }
                                    }
                                }
                                
                                // 心跳检查
                                _ = heartbeat_timer.tick() => {
                                    let elapsed = last_heartbeat.elapsed().unwrap_or_default();
                                    if elapsed > heartbeat_interval * 3 {
                                        warn!("💔 SSE 心跳超时，尝试重连");
                                        break;
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        error!("❌ SSE 连接失败: {}", e);
                        reconnect_count += 1;
                    }
                }
                
                // 等待重连
                let wait_time = std::cmp::min(
                    reconnect_interval.as_secs() * 2_u64.pow(reconnect_count.min(5)),
                    300, // 最大5分钟
                );
                
                warn!("⏰ {} 秒后尝试重新连接SSE... (第{}次重试)", wait_time, reconnect_count + 1);
                tokio::time::sleep(Duration::from_secs(wait_time)).await;
            }
            
            info!("🛑 SSE 推送监听已停止");
        });
    }
    async fn establish_sse_connection(
        endpoint: &str,
        environments: &Arc<RwLock<Vec<ConfigEnvironment>>>,
    ) -> ClientResult<SseEventStream> {
        use reqwest::Client;
        
        // 构建 SSE URL
        let sse_url = if endpoint.starts_with("http://") || endpoint.starts_with("https://") {
            format!("{}/config/events", endpoint)
        } else {
            format!("http://{}/config/events", endpoint)
        };
        
        debug!("📡 尝试连接 SSE: {}", sse_url);
        
        let client = Client::new();
        
        // 构建订阅参数
        let envs = environments.read().await;
        let env_params: Vec<String> = envs.iter()
            .map(|env| format!("{}:{}:{}", env.application, env.profile, env.label))
            .collect();
        
        let mut request = client
            .get(&sse_url)
            .header("Accept", "text/event-stream")
            .header("Cache-Control", "no-cache");
        
        // 添加订阅参数
        if !env_params.is_empty() {
            request = request.query(&[("environments", env_params.join(","))]);
        }
        
        match request.send().await {
            Ok(response) => {
                if response.status().is_success() {
                    debug!("✅ SSE 连接成功，响应状态: {}", response.status());
                    
                    // 简化实现：直接读取文本内容
                    let text = response.text().await
                        .map_err(|e| ConfigClientError::WatchError(format!("SSE读取失败: {}", e)))?;
                    
                    Ok(SseEventStream::new_from_text(text))
                } else {
                    let status = response.status();
                    let error_text = response.text().await.unwrap_or_default();
                    error!("❌ SSE 连接失败，状态: {}, 错误: {}", status, error_text);
                    Err(ConfigClientError::WatchError(format!("SSE连接失败: {} - {}", status, error_text)))
                }
            }
            Err(e) => {
                error!("❌ SSE HTTP 请求失败: {}", e);
                Err(ConfigClientError::WatchError(format!("SSE HTTP请求失败: {}", e)))
            }
        }
    }
    
    /// 接收 SSE 事件
    async fn receive_sse_event(
        event_stream: &mut SseEventStream,
    ) -> ClientResult<Option<PushMessage>> {
        match event_stream.next_event().await {
            Ok(Some(event)) => {
                match event.event_type.as_deref() {
                    Some("config_change") | Some("config_update") | Some("config_delete") => {
                        match serde_json::from_str::<PushMessage>(&event.data) {
                            Ok(push_message) => {
                                debug!("📨 收到SSE推送消息: {:?}", push_message.message_type);
                                Ok(Some(push_message))
                            }
                            Err(e) => {
                                warn!("⚠️ 解析SSE推送消息失败: {}, 消息内容: {}", e, event.data);
                                Ok(None)
                            }
                        }
                    }
                    Some("heartbeat") | Some("ping") => {
                        debug!("💓 收到SSE心跳");
                        Ok(None)
                    }
                    Some("error") => {
                        error!("❌ SSE服务器错误: {}", event.data);
                        Err(ConfigClientError::WatchError(format!("SSE服务器错误: {}", event.data)))
                    }
                    _ => {
                        debug!("📨 收到未知SSE事件类型: {:?}, 数据: {}", event.event_type, event.data);
                        Ok(None)
                    }
                }
            }
            Ok(None) => {
                debug!("📡 SSE 流已结束");
                Ok(None)
            }
            Err(e) => {
                error!("❌ SSE 事件接收失败: {}", e);
                Err(ConfigClientError::WatchError(format!("SSE事件接收失败: {}", e)))
            }
        }
    }

    /// 获取监听统计信息
    pub async fn get_stats(&self) -> WatcherStats {
        let environments = self.environments.read().await;
        let version_cache = self.version_cache.read().await;
        let config_cache = self.config_cache.read().await;

        WatcherStats {
            watched_environments: environments.len(),
            cached_versions: version_cache.len(),
            cached_configs: config_cache.len(),
            state: self.get_state().await,
        }
    }
}

/// WebSocket流类型别名
type WebSocketStream = tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;

/// 订阅消息结构
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SubscriptionMessage {
    /// 消息类型
    pub action: String,
    /// 要订阅的环境列表
    pub environments: Vec<ConfigEnvironment>,
}

/// 推送消息结构
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PushMessage {
    /// 消息类型（config_updated, config_deleted, config_created, heartbeat）
    pub message_type: String,
    /// 受影响的环境
    pub environment: ConfigEnvironment,
    /// 新的配置数据（可选）
    pub config: Option<RustCloudConfig>,
    /// 配置版本
    pub version: Option<String>,
    /// 消息时间戳
    pub timestamp: SystemTime,
    /// 附加元数据
    pub metadata: Option<HashMap<String, String>>,
}

/// SSE事件结构
#[derive(Debug, Clone)]
pub struct SseEvent {
    /// 事件类型
    pub event_type: Option<String>,
    /// 事件数据
    pub data: String,
    /// 事件ID
    pub id: Option<String>,
    /// 重连时间
    pub retry: Option<u32>,
}

/// SSE事件流
pub struct SseEventStream {
    /// 缓存的数据
    buffer: String,
    /// 事件列表
    events: Vec<SseEvent>,
    /// 当前事件索引
    current_index: usize,
}

impl SseEventStream {
    /// 创建新的SSE事件流（从字节流）
    pub fn new(byte_stream: impl Into<reqwest::Body>) -> Self {
        Self {
            buffer: String::new(),
            events: Vec::new(),
            current_index: 0,
        }
    }
    
    /// 创建新的SSE事件流（从文本）
    pub fn new_from_text(text: String) -> Self {
        let mut stream = Self {
            buffer: text,
            events: Vec::new(),
            current_index: 0,
        };
        stream.parse_all_events();
        stream
    }
    
    /// 解析所有事件
    fn parse_all_events(&mut self) {
        let mut pos = 0;
        while let Some(event_end) = self.buffer[pos..].find("\n\n") {
            let event_text = &self.buffer[pos..pos + event_end];
            if let Some(event) = self.parse_sse_event(event_text) {
                self.events.push(event);
            }
            pos += event_end + 2;
        }
    }
    
    /// 获取下一个事件
    pub async fn next_event(&mut self) -> Result<Option<SseEvent>, Box<dyn std::error::Error + Send + Sync>> {
        if self.current_index < self.events.len() {
            let event = self.events[self.current_index].clone();
            self.current_index += 1;
            Ok(Some(event))
        } else {
            Ok(None)
        }
    }
    
    /// 解析SSE事件文本
    fn parse_sse_event(&self, event_text: &str) -> Option<SseEvent> {
        let mut event_type = None;
        let mut data_lines = Vec::new();
        let mut id = None;
        let mut retry = None;
        
        for line in event_text.lines() {
            let line = line.trim();
            if line.is_empty() {
                continue;
            }
            
            if let Some((field, value)) = line.split_once(':') {
                let field = field.trim();
                let value = value.trim();
                
                match field {
                    "event" => event_type = Some(value.to_string()),
                    "data" => data_lines.push(value),
                    "id" => id = Some(value.to_string()),
                    "retry" => {
                        if let Ok(retry_value) = value.parse() {
                            retry = Some(retry_value);
                        }
                    }
                    _ => {}
                }
            } else if line.starts_with("data:") {
                // 处理无空格的data行
                data_lines.push(&line[5..]);
            }
        }
        
        if !data_lines.is_empty() {
            Some(SseEvent {
                event_type,
                data: data_lines.join("\n"),
                id,
                retry,
            })
        } else {
            None
        }
    }
}

/// 监听器统计信息
#[derive(Debug)]
pub struct WatcherStats {
    /// 监听的环境数量
    pub watched_environments: usize,
    /// 缓存的版本数量
    pub cached_versions: usize,
    /// 缓存的配置数量
    pub cached_configs: usize,
    /// 当前状态
    pub state: WatcherState,
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use tokio::sync::broadcast;

    // Mock远程配置获取器
    struct MockFetcher {
        version_counter: AtomicUsize,
    }

    impl MockFetcher {
        fn new() -> Self {
            Self {
                version_counter: AtomicUsize::new(1),
            }
        }
    }

    #[async_trait::async_trait]
    impl RemoteConfigFetcher for MockFetcher {
        async fn fetch_config(&self, _environment: &ConfigEnvironment) -> ClientResult<RustCloudConfig> {
            Ok(RustCloudConfig::default())
        }

        async fn health_check(&self) -> ClientResult<bool> {
            Ok(true)
        }

        async fn get_version(&self, _environment: &ConfigEnvironment) -> ClientResult<String> {
            let version = self.version_counter.fetch_add(1, Ordering::Relaxed);
            Ok(format!("v{}", version))
        }
    }

    #[tokio::test]
    async fn test_watcher_lifecycle() {
        let fetcher = Arc::new(MockFetcher::new());
        let (sender, _) = broadcast::channel(10);
        let strategy = WatchStrategy::Polling {
            interval: Duration::from_millis(100),
            version_check: true,
        };

        let watcher = ConfigWatcher::new(fetcher, strategy, sender);

        // 测试初始状态
        assert_eq!(watcher.get_state().await, WatcherState::Stopped);

        // 添加监听环境
        let environment = ConfigEnvironment {
            application: "test-app".to_string(),
            profile: "dev".to_string(),
            label: "main".to_string(),
            version: None,
            state: None,
        };
        watcher.add_environment(environment).await;

        // 启动监听
        watcher.start().await.unwrap();
        assert_eq!(watcher.get_state().await, WatcherState::Running);

        // 暂停监听
        watcher.pause().await;
        assert_eq!(watcher.get_state().await, WatcherState::Paused);

        // 恢复监听
        watcher.resume().await;
        assert_eq!(watcher.get_state().await, WatcherState::Running);

        // 停止监听
        watcher.stop().await;
        assert_eq!(watcher.get_state().await, WatcherState::Stopped);
    }

    #[tokio::test]
    async fn test_environment_management() {
        let fetcher = Arc::new(MockFetcher::new());
        let (sender, _) = broadcast::channel(10);
        let strategy = WatchStrategy::default();

        let watcher = ConfigWatcher::new(fetcher, strategy, sender);

        let env1 = ConfigEnvironment {
            application: "app1".to_string(),
            profile: "dev".to_string(),
            label: "main".to_string(),
            version: None,
            state: None,
        };

        let env2 = ConfigEnvironment {
            application: "app2".to_string(),
            profile: "prod".to_string(),
            label: "main".to_string(),
            version: None,
            state: None,
        };

        // 添加环境
        watcher.add_environment(env1.clone()).await;
        watcher.add_environment(env2.clone()).await;

        let stats = watcher.get_stats().await;
        assert_eq!(stats.watched_environments, 2);

        // 移除环境
        watcher.remove_environment(&env1).await;

        let stats = watcher.get_stats().await;
        assert_eq!(stats.watched_environments, 1);
    }
}